package org.caw.hbase.explorer.impl;

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.caw.exception.CawException;
import org.caw.exception.CawNoRecordFoundException;
import org.caw.hbase.HBaseTablePool;
import org.caw.hbase.explorer.DeleteManager;
import org.caw.util.ConvertUtil;

public class HBaseDeleteManagerImpl implements DeleteManager {

	private HBaseTablePool pool;

	@Override
	public void delete(String tableName, String rowkey) throws CawException {
		delete(tableName, rowkey, null, null);
	}

	@Override
	public void delete(String tableName, String rowkey, Long timestamp) throws CawException {
		delete(tableName, rowkey, null, timestamp);
	}

	@Override
	public void delete(String tableName, String rowkey, List<String> columns) throws CawException {
		delete(tableName, rowkey, columns, null);
	}

	@Override
	public void delete(String tableName, String rowkey, List<String> columns, Long timestamp) throws CawException {
		HTableInterface table = null;

		try {
			table = pool.getTable(tableName);
			Get get = ConvertUtil.buildGet(rowkey, null, null, timestamp);
			if (table.exists(get)) {
				Delete delete = ConvertUtil.buildDelete(rowkey, columns, timestamp);
				table.delete(delete);
			} else {
				throw new CawNoRecordFoundException("Record with rowkey " + rowkey + "doesn't exist in table " + tableName);
			}
		} catch (IOException e) {
			throw new CawException("Failed to delete record", e);
		} finally {
			if (table != null) {
				pool.putTable(table);
			}
		}
	}

	public HBaseTablePool getPool() {
		return pool;
	}

	public void setPool(HBaseTablePool pool) {
		this.pool = pool;
	}

}
